package com.simba.server.framedecoder;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;

import com.simba.server.framedecoder.decoders.*;
import org.apache.log4j.Logger;
import org.jboss.netty.channel.Channel;
import com.simba.common.RequestHandler;
import com.simba.common.decoder.Decoder;
import com.simba.common.decoder.DecoderContainer;
import com.simba.common.decoder.DefaultDecoderContainer;
import com.simba.common.frameserver.FrameChannelManager;
import com.simba.common.log.LogUtil;
import com.simba.common.thread.CommonExecutorFactory;
import com.simba.common.thread.CounterThreadFactory;
import com.simba.server.ServerParam;
import com.simba.server.components.SortCommonDefines;
import com.simba.server.components.SorterParams;

public class FrameDecoderManager implements RequestHandler
{
	private static class FrameDecoderManagerHolder
	{
		private static final FrameDecoderManager INSTANCE = new FrameDecoderManager();
	}

	public static final byte CONNECT_MESSGAE_END = (byte) 0xee;
	public static final byte CONNECT_MESSGAE_HEAD = (byte) 0xff;
	protected final static Logger logger = Logger.getLogger(FrameDecoderManager.class);
	private static int ASYNC_EXECUTOR_POOL_SIZE = 8;
	private static final String ASYNC_OPERATION_EXECUTOR_NAME = "Frame-Decode-Task";

	public static FrameDecoderManager getInstance()
	{
		return FrameDecoderManagerHolder.INSTANCE;
	}

	private Executor asyncDecodeExecutor;
	private BlockingQueue<Runnable> asyncDecodeOperationQueue;
	private final DecoderContainer decoderContainer = new DefaultDecoderContainer();

	private FrameDecoderManager()
	{

	}

	@Override
	public void handleRequest(final Channel channel, final Object message)
	{
		Runnable executeRunnable = new Runnable()
		{
			@Override
			public void run()
			{
				try
				{
					if (!(message instanceof byte[]))
					{
						LogUtil.LogDebug(FrameDecoderManager.logger, "FrameDecoder only can decode byte frame,but message is not instance of byte[].");

						FrameChannelManager.getInstance().commitResponse(channel, SortCommonDefines.ERROR_MESSAGE, true);
						return;
					}

					final byte[] content = (byte[]) message;
					if (content[0] != 0x01)
					{
						FrameChannelManager.getInstance().commitResponse(channel, SortCommonDefines.ERROR_MESSAGE, true);
						return;
					}

					byte command = content[SortCommonDefines.PROTOCOL_COMMAND_POSITION];
					Decoder frameDecoder = FrameDecoderManager.this.getDecoderByCommandNum(command);
					if (frameDecoder != null)
					{
						frameDecoder.doDecode(channel, message);
					}
				}
				catch (Exception ex)
				{
					LogUtil.LogError(FrameDecoderManager.logger,
							"FrameDecoderManager catch exception : " + ex.getMessage());
					// 让调用栈能继续处理线程的终端状态
					if (ex instanceof InterruptedException)
					{
						Thread.currentThread().interrupt();
					}
				}
			}
		};

		if (SorterParams.getInstance().isNewThreadEachRequest())
		{
			this.executeInNewThread(executeRunnable);
		}
		else
		{
			this.executeInThreadPool(executeRunnable);
		}
	}

	public void init()
	{
		this.initAsyncTaskExcutor();
		this.initDecoderContainer();
	}

	public void postInitProcess()
	{
		String requestMode = "Request In New Thread";
		if (!SorterParams.getInstance().isNewThreadEachRequest())
		{
			requestMode = "Request In Thread Pool";
		}
		LogUtil.LogDebug(FrameDecoderManager.logger,
				"FrameDecoderManager initialize successfully, Pool size:["
						+ FrameDecoderManager.ASYNC_EXECUTOR_POOL_SIZE + "] reqeust mode:["
						+ requestMode + "]");
	}

	private void executeInNewThread(final Runnable executeRunnable)
	{
		new Thread(executeRunnable).start();
	}

	private void executeInThreadPool(final Runnable executeRunnable)
	{
		/** *********** identify init command begin *********** */

		this.runAsyncDecodeTask(executeRunnable);
	}

	private Decoder getDecoderByCommandNum(byte command)
	{
		Decoder frameDecoder = null;
		switch (command)
		{
			case SortCommonDefines.COMMAND_0X02_WHEEL_PACKET_ON:
				frameDecoder = this.decoderContainer.getDecoder(PacketOnDecoder.NAME);
				break;
			case SortCommonDefines.COMMAND_0X05_WHEEL_PACKET_OFF:
				frameDecoder = this.decoderContainer.getDecoder(PacketOffDecoder.NAME);
				break;
			case SortCommonDefines.COMMAND_0X70_WHEEL_KEEP_ALIVE:
				frameDecoder = this.decoderContainer.getDecoder(KeepAliveDecoder.NAME);
				break;
			case SortCommonDefines.COMMAND_0X07_WHEEL_SLOT_PACK:
				frameDecoder = this.decoderContainer.getDecoder(SlotPackDecoder.NAME);
				break;
			default:
				break;
		}
		return frameDecoder;
	}

	private void initAsyncTaskExcutor()
	{
		if (ServerParam.DECODER_THREAD_POOL_SIZE > 0)
		{
			FrameDecoderManager.ASYNC_EXECUTOR_POOL_SIZE = ServerParam.DECODER_THREAD_POOL_SIZE;
		}
		this.asyncDecodeOperationQueue = new LinkedBlockingQueue<Runnable>();
		this.asyncDecodeExecutor = CommonExecutorFactory.newFixedThreadPool(
				FrameDecoderManager.ASYNC_EXECUTOR_POOL_SIZE,
				new CounterThreadFactory(FrameDecoderManager.ASYNC_OPERATION_EXECUTOR_NAME),
				this.asyncDecodeOperationQueue);
	}

	private void initDecoderContainer()
	{
		this.decoderContainer.addDecoder(KeepAliveDecoder.NAME, KeepAliveDecoder.getInstance());
		this.decoderContainer.addDecoder(PacketOnDecoder.NAME, PacketOnDecoder.getDecoderInstance());
		this.decoderContainer.addDecoder(PacketOffDecoder.NAME, PacketOffDecoder.getDecoderInstance());
		this.decoderContainer.addDecoder(SlotPackDecoder.NAME, SlotPackDecoder.getDecoderInstance());
	}

	private void runAsyncDecodeTask(Runnable task)
	{
		try
		{
			if (task != null)
			{
				this.asyncDecodeExecutor.execute(task);
			}
		}
		catch (Exception e)
		{
			LogUtil.LogError(FrameDecoderManager.logger,
					"Run async task to decode failed : " + e.getMessage());
		}
	}
}
